/*  File Name	: TimerHeap.cpp
 *  Author		: Javie Pong
 *	 E-mail		: 823960168@qq.com
 *  Created on	: 2012-5-12
 *  Description:
 */

#include <string.h>
#include "TimerHeap.h"
#include "Lock.h"
#include "ReactorImpl.h"
#include "SocketPair.h"

void TimerNode::reset(void)
{
	this->event_handler_ = NULL;
	this->interval_ = 0;
	this->next_ticker_ = 0;
	this->is_suspend_ = true;
	this->is_pop = false;
	this->is_cancel = false;
	this->is_refresh = false;
	this->timer_id_ = 0;
	this->heap_index_ = -1;
}

TimerHeap::TimerHeap(ReactorImpl *reactor)
{
	this->reactor_ = reactor;
    this->heap_ = NULL;
    this->max_size_ = 0;
    this->cur_size_ = 0;
    this->is_open_ = false;
    this->timer_id_ = 1;
    this->node_map_.clear();

    SocketPair::socketpair_fd_reset(this->timer_notify_fd);
}

TimerHeap::~TimerHeap()
{
    if( NULL!=this->heap_ )
    {
        ::free(this->heap_);
        this->heap_ = NULL;
    }
    this->reactor_ = NULL;
    this->node_map_.clear();
    this->node_pool_.recycle();

    this->close_notify_fd();
}

//void TimerHeap::set_reactor(ReactorImpl *reactor)
//{
//	this->reactor_ = reactor;
//}

int TimerHeap::open(int max_size)
{
    if( this->is_open_ )
    {
        return -1;
    }
    max_size = max_size < 0? 2048:max_size;
    this->max_size_ = max_size;

    this->heap_ = (TimerNode**)malloc(sizeof(TimerNode*)*this->max_size_);

    if(NULL == this->heap_ )
    {
        return -1;
    }
    this->is_open_ = true;

    this->node_pool_.set_pool_size(this->max_size_*3);
    this->node_pool_.set_pool_name("TimerNode");

    if(this->init_notify_fd()!=0)
    {
    	LOG((ERROR_LOG,"TimerHeap::open->init_notify_fd error"));
        if( NULL!=this->heap_ )
        {
            ::free(this->heap_);
            this->heap_ = NULL;
        }
        return -1;
    }
    return 0;
}

long TimerHeap::register_timer(EventHandler *eh,TimeValue delay,TimeValue interval)
{
	if(NULL==eh)
	{
		return -1;
	}

	TimerNode *node = NULL;
	if(this->node_pool_.pop(node)!=0)
	{
		return -1;
	}
	node->event_handler_ = eh;
	node->next_ticker_ = TimeValue::now_time_in_msec() + delay.sec()*1000+delay.msec();
	node->interval_ = interval.sec()*1000+interval.msec();
	node->is_suspend_ = false;
	node->timer_id_ = this->generate_timer_id();

	if(this->heap_push(node)!=0)
	{
		if(node!=NULL)
		{
			this->node_pool_.push(node);
		}
		return -1;
	}

	LockGuardReturn(RT_MUTEX, lock, this->rt_mutex_, -1);
	TimerNodeMap::iterator iter = this->node_map_.find(node->timer_id_);
	if (iter == this->node_map_.end())
	{
		this->node_map_.insert(std::pair<int,
				TimerNode*>(node->timer_id_,node));
	}
	else
	{
		iter->second->event_handler_ = eh;
	}

	char buf[1]={0};
	SocketPair::socketpair_notify(this->timer_notify_fd[1],buf,1);

   return node->timer_id_;
}

int TimerHeap::suspend_timer(long timer_id)
{
	LockGuardReturn(RT_MUTEX, lock, this->rt_mutex_, -1);
	TimerNodeMap::iterator iter = this->node_map_.find(timer_id);
	if (iter != this->node_map_.end())
	{
		iter->second->is_suspend_ = true;
		return 0;
	}
	return -1;
}

int TimerHeap::resume_timer(long timer_id)
{
	LockGuardReturn(RT_MUTEX, lock, this->rt_mutex_, -1);
	TimerNodeMap::iterator iter = this->node_map_.find(timer_id);
	if (iter != this->node_map_.end())
	{
		iter->second->is_suspend_ = false;
		return 0;
	}
	return -1;
}

int TimerHeap::refresh_timer(long timer_id,TimeValue new_time)
{
	LockGuardReturn(RT_MUTEX, lock, this->rt_mutex_, -1);
	TimerNodeMap::iterator iter = this->node_map_.find(timer_id);
	if (iter != this->node_map_.end())
	{
		iter->second->interval_ = new_time.sec()*1000+new_time.msec();
		iter->second->next_ticker_ = TimeValue::now_time_in_msec() + iter->second->interval_;
		iter->second->is_refresh = true;
		if(!iter->second->is_pop)
		{
			int parent = (iter->second->heap_index_-1)/2;
			if(this->heap_[parent]->next_ticker_ > iter->second->next_ticker_)
			{
				this->heap_up_sort(iter->second->heap_index_);
			}
			else
			{
				this->heap_down_sort(iter->second->heap_index_);
			}
		}
		char buf[1]={0};
		SocketPair::socketpair_notify(this->timer_notify_fd[1],buf,1);
		return 0;
	}
	return -1;
}

int TimerHeap::remove_timer(long timer_id)
{
	if(timer_id==-1)
	{
		return 0;
	}
	LockGuardReturn(RT_MUTEX,lock,this->rt_mutex_,-1);
	TimerNodeMap::iterator iter=this->node_map_.find(timer_id);
	if(iter!=this->node_map_.end())
	{
		iter->second->is_cancel = true;
		if(!iter->second->is_pop)
		{
			this->heap_earse(iter->second);
			this->pool_push(iter->second);
			char buf[1]={0};
			SocketPair::socketpair_notify(this->timer_notify_fd[1],buf,1);
		}
		this->node_map_.erase(timer_id);
		return 0;
	}
    return -1;
}

TimerNode* TimerHeap::get_heap_top(void)
{
	LockGuardReturn(RT_MUTEX,lock,this->rt_mutex_,NULL);
	return this->heap_[0];
}

long TimerHeap::generate_timer_id(void)
{
	LockGuardReturn(RT_MUTEX,lock,this->rt_mutex_,-1);
    return this->timer_id_++;
}

int TimerHeap::next_timeout_value(void)
{
	LockGuardReturn(RT_MUTEX,lock,this->rt_mutex_,-1);
	if(this->heap_[0]!=NULL)
	{
		int time = 0;
		long now = TimeValue::now_time_in_msec();
		time = this->heap_[0]->next_ticker_-TimeValue::now_time_in_msec();
		//LOG((DEBUG_LOG,"TimerHeap::next_timeout_value[id:%d][%L][%L][%d]",
		//		this->heap_[0]->timer_id_,this->heap_[0]->next_ticker_,now,time));
		return time;
	}
	return 100;
}

int TimerHeap::timer_count(void)
{
	LockGuardReturn(RT_MUTEX,lock,this->rt_mutex_,-1);
	return this->cur_size_;
}

int TimerHeap::heap_pop(TimerNode *&node)
{
	node = NULL;
	LockGuardReturn(RT_MUTEX, lock, this->rt_mutex_, -1);
	if(this->cur_size_==0)
	{
		return 0;
	}
	bool is_suspend=false;
	do
	{
		TimerNode *tmp = this->heap_[0];
		this->heap_[0] = this->heap_[this->cur_size_ - 1];
		this->heap_[this->cur_size_ - 1] = NULL;
		this->cur_size_--;
		//LOG((DEBUG_LOG,"TimerHeap::heap_pop() current size:[%d]",this->cur_size_));
		if(this->cur_size_>0)
		{
			this->heap_down_sort();
		}
		if(tmp->is_suspend_)
		{
			tmp->next_ticker_ = TimeValue::now_time_in_msec()+tmp->interval_;
			this->heap_push(tmp);
			is_suspend = true;
			continue;
		}
		is_suspend = false;
		node = tmp;
	}
	while (is_suspend && this->cur_size_>0);

	if(node!=NULL)
	{
		node->is_pop = true;
	}
	return 0;
}

int TimerHeap::heap_push(TimerNode *node)
{
	LockGuardReturn(RT_MUTEX,lock,this->rt_mutex_,-1);

	if(NULL==node)
	{
		return -1;
	}
	if(this->cur_size_+1>=this->max_size_ && this->growth()!=0)
	{
		return -1;
	}
	this->heap_[this->cur_size_]= node;
	this->heap_[this->cur_size_]->is_pop = false;
	this->cur_size_++;
	this->heap_up_sort(this->cur_size_-1);

	//LOG((DEBUG_LOG,"TimerHeap::heap_push() current size:[%d]",this->cur_size_));
//	for(int i=0;i<this->cur_size_;i++)
//	{
//		LOG((DEBUG_LOG,"TimerHeap::heap_push()[%L][%L]",
//				this->heap_[i]->timer_id_,this->heap_[i]->next_ticker_));
//	}

	return 0;
}

void TimerHeap::heap_up_sort(int up_node)
{
	TimerNode *tmp=this->heap_[up_node];

	if(up_node <0)
	{
		return;
	}
	if(up_node==0)
	{
		this->heap_[up_node]->heap_index_ = up_node;
		return;
	}

	int parent = (up_node-1)/2;
	do
	{
		if(this->heap_[parent]->next_ticker_ > tmp->next_ticker_)
		{
			this->heap_[up_node]= this->heap_[parent];
			this->heap_[up_node]->heap_index_ = up_node;
			up_node = parent;
			parent = (up_node-1)/2;
			continue;
		}
		break;
	}
	while(up_node>0);

	this->heap_[up_node]=tmp;
	this->heap_[up_node]->heap_index_ = up_node;
}

void TimerHeap::heap_down_sort(int down_node)
{
	int child = 2*down_node+1;
	TimerNode *tmp = this->heap_[down_node];
	while(child< this->cur_size_)
	{
		if(child+1 < this->cur_size_ &&
				this->heap_[child]->next_ticker_ > this->heap_[child+1]->next_ticker_)
		{
			child++;
		}
		if(this->heap_[child]->next_ticker_ < tmp->next_ticker_)
		{
			this->heap_[down_node] = this->heap_[child];
			this->heap_[down_node]->heap_index_ = down_node;
			down_node = child;
			child = 2*down_node +1;
			continue;
		}
		break;
	}
	this->heap_[down_node] = tmp;
	this->heap_[down_node]->heap_index_ = down_node;
}

int TimerHeap::growth(int size)
{
	this->heap_=(TimerNode**)::realloc(this->heap_,(this->max_size_+size)*sizeof(TimerNode*));
	if(NULL==this->heap_)
	{
		LOG((ERROR_LOG,"TimerHeap::growth() failed cur_size:[%d] max_size:[%d]\n",
						this->cur_size_,this->max_size_));
		return -1;
	}
	this->max_size_ +=size;
	LOG((INFO_LOG,"TimerHeap::growth() current size:[%d] growth:[%d]",this->max_size_,size));
	return 0;
}

void TimerHeap::heap_earse(TimerNode *node)
{
	if(NULL==node)
	{
		return ;
	}

	int index = node->heap_index_;
	int parent = (index-1)/2;
	int last = this->cur_size_-1;
	if(index<0)
	{
		return;
	}
	if(node->is_pop==false)
	{
		this->cur_size_--;
	}

	if(index==0 && this->cur_size_<=0)
	{
		this->heap_[index] = NULL;
		return ;
	}

	this->heap_[index] = this->heap_[last];
	if(this->heap_[index]->next_ticker_ < this->heap_[parent]->next_ticker_)
	{
		this->heap_up_sort(index);
	}
	else
	{
		this->heap_down_sort(index);
	}

	//LOG((INFO_LOG,"TimerHeap::heap_earse() current size:[%d]",this->cur_size_));
}

int TimerHeap::pool_push(TimerNode *node)
{
	if(NULL==node)
	{
		return -1;
	}
	if(this->node_map_.find(node->timer_id_)!=this->node_map_.end())
	{
		this->node_map_.erase(node->timer_id_);
	}
	return this->node_pool_.push(node);
}

int TimerHeap::init_notify_fd(void)
{
	if(NULL==this->reactor_)
	{
		return -1;
	}

	int result = 0;
	if (this->timer_notify_fd[0] < 0)
	{
		result = SocketPair::socketpair(this->timer_notify_fd);
		if (result != 0)
		{
			return result;
		}
	}

	if (this->timer_notify_fd[0] > 0)
	{
		SocketPair::socketpair_set_nonblock(this->timer_notify_fd[0]);
		SocketPair::socketpair_set_reuse(this->timer_notify_fd[0]);
	}
	if (this->timer_notify_fd[1] > 0)
	{
		SocketPair::socketpair_set_nonblock(this->timer_notify_fd[1]);
		SocketPair::socketpair_set_reuse(this->timer_notify_fd[1]);
	}

	//register to reactor
	if(this->reactor_->register_handler(this->timer_notify_fd[0],&this->handler_,
			EventHandler::READ_MASK)!=0)
	{
		this->close_notify_fd();
		return -1;
	}

	return 0;
}

void TimerHeap::close_notify_fd(void)
{
	SocketPair::socketpair_close(this->timer_notify_fd[0]);
	SocketPair::socketpair_close(this->timer_notify_fd[1]);
	SocketPair::socketpair_fd_reset(this->timer_notify_fd);
}

int TimerHeap::handle_event(void)
{
	TimerNode *timer = NULL;
	while (this->next_timeout_value() <= 0)
	{
		this->heap_pop(timer);

		if (NULL == timer)
		{
			return 0;
		}
		else
		{
			int result = 0;
			if (timer->is_cancel == false)
			{
				result = timer->event_handler_->handle_timeout();
				if (result != 0)
				{
					LockGuardReturn(RT_MUTEX, lock, this->rt_mutex_, -1);
					timer->is_cancel = true;
				}
			}

			if (timer->interval_ > 0 && timer->is_cancel == false)
			{
				{
					LockGuardReturn(RT_MUTEX, lock, this->rt_mutex_, -1);
					if (!timer->is_refresh)
					{
						timer->next_ticker_ += timer->interval_;
					}
					timer->is_refresh = false;
				}
				this->heap_push(timer);
			}
			else
			{
				this->remove_timer(timer->timer_id_);
				this->pool_push(timer);
			}
		}
	}
	return 0;
}
